package com.oildatacollection.summary;

import com.oildatacollection.database.InfluxdbConnection;
import com.oildatacollection.util.YamlLoader;
import com.rpc.OilInfoGet;
import javafx.util.Pair;
import lombok.extern.slf4j.Slf4j;
import org.influxdb.InfluxDBIOException;
import org.influxdb.dto.QueryResult;
import java.net.MalformedURLException;
import java.rmi.Naming;
import java.rmi.NotBoundException;
import java.rmi.RemoteException;
import java.util.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.stream.Collectors;

@Slf4j
public class DataSummaryInfluxImpl implements DataSummary {
    private final Timer timer;
    private final Lock lock;
    private final Condition cond;
    private boolean changed;
    private final AtomicBoolean stop;
    private final InfluxdbConnection influxConn;
    private Thread summaryThread;
    private OilInfoGet oilInfo;

    public DataSummaryInfluxImpl() {
        timer = new Timer();
        lock = new ReentrantLock();
        cond = lock.newCondition();
        changed = false;
        stop = new AtomicBoolean(false);
        influxConn = new InfluxdbConnection("dataSummary");
        summaryThread = null;
        oilInfo = null;
    }

    private void doThread() {
        while (!stop.get()) {
            try {
                // 重置任务
                TimerTask timerTask = new TimerTask() {
                    @Override
                    public void run() {
                        doTask();
                        cancel();
                    }
                };
                // 获取汇总的间隔 毫秒为单位
                long curInterval = getInterval() * 1000;
                timer.schedule(timerTask, curInterval);

                lock.lock();
                while (!changed) {
                    try {
                        cond.await();
                    } catch (InterruptedException e) {
                        log.error("DataSummaryInfluxImpl:cond错误", e);
                    }
                }
            } finally {
                changed = false;
                lock.unlock();
            }
        }
    }

    private void initConfig() {
        if (summaryThread != null) {
            try {
                summaryThread.join();
            } catch (InterruptedException e) {
                log.error("summaryThread错误", e);
                return;
            }
        }

        // 连接数据库
        try {
            influxConn.influxDbBuild();
        } catch (InfluxDBIOException e) {
            log.error("汇总模块--dataSummary: 数据库连接失败", e);
            return;
        }
        log.trace("汇总模块--dataSummary: 数据库连接成功");

        // 初始化rpc模块
        YamlLoader yaml = new YamlLoader("app-config/application-rpc.yml");
        Map<String, Object> clientObj = (Map<String, Object>)yaml.get("client");
        String name = (String) clientObj.get("name");
        try {
            oilInfo = (OilInfoGet) Naming.lookup(name);
        } catch (RemoteException | NotBoundException | MalformedURLException  e) {
            log.error("dataSummary: RPC连接失败", e);
            return;
        }
        log.trace("汇总模块初始化完毕");
    }

    @Override
    public void startSummary() {
        initConfig();
        summaryThread = new Thread(new Runnable() {
            @Override
            public void run() {
                doThread();
            }
        });
        summaryThread.start();
    }

    @Override
    public void resetSummary() {
        lock.lock();
        changed = true;
        cond.signalAll();
        lock.unlock();
    }

    @Override
    public void stopSummary() {
        stop.set(true);
        resetSummary();
    }

    @Override
    public void waitForEnd() {
        if (summaryThread != null) {
            try {
                summaryThread.join();
            } catch (InterruptedException e) {
                log.error("summaryThread", e);
            }
        }
    }

    /**
     * 获取汇总的时间间隔(rpc)
     * @return
     */
    private long getInterval() {
        long ret = 0;
        try {
            ret = oilInfo.getSummaryInterval();
        } catch (RemoteException e) {
            log.error("RPC调用有误", e);
        }
        return ret;
    }

    /**
     * 获取需要汇总的井口id (rpc)
     * @return
     */
    private List<String> getWellIds() {
        List<String> ret = null;
        try {
            ret = oilInfo.getWellIds();
        } catch (RemoteException e) {
            log.error("RPC调用有误", e);
        }
        return ret;
    }

    /**
     * 获取上次汇总的时间 (rpc)
     * @param wellID
     * @return
     */
    private String getLastSummaryTime(String wellID) {
        String retTime = "";
        try {
            retTime = oilInfo.getLastSummaryTime(wellID);
        } catch (RemoteException e) {
            log.error("RPC调用有误", e);
        }
        return retTime;
    }

    private Pair<List<String>, List<Object>> getOneResult(String sql) {
        QueryResult.Result resultOne = influxConn.query(sql).getResults().get(0);
        if (resultOne.getSeries() == null) {
            return null;
        }
        List<String> columns = resultOne.getSeries().stream()
                .map(QueryResult.Series::getColumns).collect(Collectors.toList()).get(0);
        List<Object> values = resultOne.getSeries().stream()
                .map(QueryResult.Series::getValues).collect(Collectors.toList()).get(0).get(0);
        return new Pair<>(columns, values);
    }

    /**
     * 获取每个井口汇总的时间范围 map中
     * @return
     */
    private Map<String, Pair<String, String>> getSummaryRange() {
        Map<String, Pair<String, String>> ret = new HashMap<>();
        // 获取需要汇总的井口
        List<String> wells = getWellIds();
        String tableName = influxConn.getUseTableName();
        // 获取每个井口的最新汇总时间
        for (String wellID : wells) {
            String sql = "select * from " + tableName + " where wellID='" + wellID + "' order by time desc limit 1;";
            Pair<List<String>, List<Object>> res = getOneResult(sql);
            if (res == null) {
                continue;
            }
            List<String> columns = res.getKey();
            List<Object> values = res.getValue();
            String latestTime = "";
            for (int idx = 0; idx < columns.size(); idx++) {
                if (columns.get(idx).equals("time")) {
                    latestTime = values.get(idx).toString();
                    break;
                }
            }
            // 获取上次合并的时间
            String lastTime = getLastSummaryTime(wellID);
            ret.put(wellID, new Pair<>(lastTime, latestTime));
        }
        return ret;
    }

    private void doTask() {
        try {
            // 获取汇总的范围
            Map<String, Pair<String, String>> summaryRang = getSummaryRange();
            if (summaryRang.size() > 0) {
                String tableName = influxConn.getUseTableName();
                for (String wellID : summaryRang.keySet()) {
                    Pair<String, String> ret = summaryRang.get(wellID);
                    String lastTime = ret.getKey();
                    String latestTime = ret.getValue();
                    String sql = "select sum(flow) as sumFlow, " +
                            "mean(temperature) as meanTemp, " +
                            "mean(pressure) as meanPress, " +
                            "mean(waterContent) as meanWaterCont, " +
                            "sum(oilContent) as sumOilCont, " +
                            "mean(gasContent) as meanGasCont from " + tableName +
                            " where wellID='" + wellID +
                            "' and time > '" + lastTime + "' and time <= '" + latestTime + "';";
                    // 执行sql
                    Pair<List<String>, List<Object>> res = getOneResult(sql);
                    if (res == null) {
                        continue;
                    }
                    List<String> columns = res.getKey();
                    List<Object> values = res.getValue();
                    if (columns.size() != values.size()) {
                        log.error("汇总结果有误");
                        return;
                    }
                    Map<String, Object> summaryRes = new HashMap<>();
                    // 写入执行结果
                    for (int i = 0; i < columns.size(); i++) {
                        String colName = columns.get(i);
                        Object value = values.get(i);
                        summaryRes.put(colName, value);
                    }
                    // 写入汇总时间 井口ID
                    summaryRes.put("latestTime", latestTime);
                    summaryRes.put("wellID", wellID);
                    // 将结果发送至另一模块
                    oilInfo.sendSummaryResult(summaryRes);
                }
            }
        } catch (Exception e) {
            log.error("RPC调用失败", e);
        } finally {
            // 执行完任务后通知主线程重置任务
            lock.lock();
            changed = true;
            cond.signalAll();
            lock.unlock();
        }
    }
}
